package configcenter.zookeeper.primary.server;

import common.constants.CommonConstants;
import common.constants.RegistryConstants;
import common.model.MagicUrl;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ZookeeperServer implements RegistryServer {

  private static final Logger LOG = LoggerFactory.getLogger(ZookeeperServer.class);
  private ZooKeeper zookeeper = null;
  private MagicUrl url;
  private volatile boolean isAvailable = false;

  public ZookeeperServer(final MagicUrl url) {
    this.url = url;
  }

  public void init() {
    String env = CommonConstants.MAGIC;
    String service = url.getParameter(RegistryConstants.REGISTRY_SERVICE);
    int port = url.getParameter(CommonConstants.BIND_PORT_KEY, url.getPort());
    String zkpath = url.getParameter(RegistryConstants.REGISTRY_CONNECT_PARH);
    CountDownLatch c = new CountDownLatch(1);
    if (zookeeper == null) {
      try {
        zookeeper = new ZooKeeper(zkpath, RegistryConstants.SESSION_TIMEOUT, new ZookeeperWatcher(c, this));
          isAvailable = true;
      } catch (IOException e) {
        LOG.error("zk server faild service:" + env + "-" + service, e);
      }
    }

    try {
      int retry = 0;
      boolean connected = false;
      while (retry++ < RegistryConstants.RETRY_TIMES) {
        if (c.await(5, TimeUnit.SECONDS)) {
          connected = true;
          break;
        }
      }
      if (!connected) {
        LOG.error("zk connected fail! :" + env + "-" + service);
        isAvailable = false;
        throw new IllegalArgumentException("zk connected fail!");
      }
    } catch (InterruptedException e) {
      LOG.error(e.getMessage(), e);
    }

    if (zookeeper == null) {
      LOG.error("zk server is null service:" + env + "-" + service);
      isAvailable = false;
      throw new IllegalArgumentException("zk server can't be null");
    }
    try {
      String envPath = env.startsWith("/") ? env : "/".concat(env);
      if (zookeeper.exists(envPath, null) == null) {
        zookeeper
            .create(envPath, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      }
      String servicePath = service.startsWith("/") ? service : "/".concat(service);

      if (zookeeper.exists(envPath + servicePath, null) == null) {
        zookeeper.create(envPath + servicePath, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);
      }
      String ip = url.getParameter(CommonConstants.BIND_IP_KEY, url.getHost());
      if (StringUtils.isEmpty(ip)) {
        throw new IllegalArgumentException("ip can't be null");
      }
      String childpath;
      String serviceType = "/".concat(CommonConstants.PROVIDER);
      String ess;
      if (zookeeper.exists(ess = (envPath + servicePath + serviceType), null) == null) {
        zookeeper
            .create(ess, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      }
      if (zookeeper.exists(childpath = (ess + "/" + ip + ":" + port), null)
          == null) {
        zookeeper.create(childpath, url.toFullString().getBytes(RegistryConstants.UTF_8), ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL);
      }
      LOG.info("zk server init success,info url={}", url.toFullString());
    } catch (KeeperException e) {
      LOG.error(e.getMessage(), e);
    } catch (InterruptedException e) {
      LOG.error(e.getMessage(), e);
    } catch (UnsupportedEncodingException e) {
      LOG.error(e.getMessage(), e);
    }
  }

  public void destroy() {
    if (zookeeper != null) {
      try {
        zookeeper.close();
        zookeeper = null;
      } catch (InterruptedException e) {
        LOG.error("the service 【{}】zk close faild info={}", url.toFullString());
      }
    }
  }

  @Override
  public MagicUrl getUrl() {
      return url;
  }

  @Override
  public boolean isAvailable() {
      return isAvailable;
  }

  public void setAvailable(boolean available) {
      isAvailable = available;
  }
}
